package com.gitee.lirenqing.flinksimpledatamysqlsource;

import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

public class MysqlRichSource extends RichSourceFunction<String> {
    private volatile boolean running = true;
    private HikariDataSource ds = null;
    private Connection conn = null;
    private Statement statement = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        HikariConfig hikariConfig = new HikariConfig();
        //设置url
        hikariConfig.setJdbcUrl("jdbc:mysql://localhost:3306/mysql?characterEncoding=UTF-8&&autoReconnect=true&failOverReadOnly=false&zeroDateTimeBehavior=CONVERT_TO_NULL&useSSL=false&serverTimezone=Asia/Shanghai");
        //数据库帐号
        hikariConfig.setUsername("root");
        //数据库密码
        hikariConfig.setPassword("123456");
        hikariConfig.setDriverClassName("com.mysql.cj.jdbc.Driver");
        hikariConfig.addDataSourceProperty("cachePrepStmts", "true");
        hikariConfig.addDataSourceProperty("prepStmtCacheSize", "250");
        hikariConfig.addDataSourceProperty("prepStmtCacheSqlLimit", "2048");
        ds = new HikariDataSource(hikariConfig);
    }

    @Override
    public void run(SourceContext<String> sourceContext) throws Exception {

        ResultSet rs = null;
        Connection conn = ds.getConnection();
        statement = conn.createStatement();
        rs = statement.executeQuery("select * from user");
        while (running && rs.next()) {
            System.out.println(rs.getString("User"));
            System.out.println(rs.getString("Host"));
            sourceContext.collect("user:" + rs.getString("User") + " host:" + rs.getString("Host"));
        }
    }

    @Override
    public void cancel() {
        running = false;

        if (statement != null) {
            try {
                statement.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }

        if (conn != null) {
            try {
                conn.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }

    }

    @Override
    public void close() throws Exception {
        super.close();
        if (statement != null) {
            try {
                statement.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }

        if (conn != null) {
            try {
                conn.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }
}
